package main

import (
	"bytes"
	"sync"
	"time"

	"../logger"
	"../util"
	cluster "github.com/bsm/sarama-cluster"
)

type MsgBuffer struct {
	buff *bytes.Buffer
	mu   sync.Mutex
}

func (logic *Logic) consumeMsg() error {
	config := cluster.NewConfig()
	config.Consumer.Return.Errors = true
	// config.Group.Return.Notifications = true

	brokers := []string{"127.0.0.1:9092"}
	topics := []string{"test"}

	consumer, err := cluster.NewConsumer(brokers, "my-consumer-group", topics, config)
	if err != nil {
		return err
	}
	defer CloseConsumer(consumer)

	after_1_minute := time.After(30 * time.Second) //1 * time.Minute)
	for {
		select {
		case err := <-consumer.Errors():
			if err != nil {
				logger.Errorf("Fail to consume message from Kafka: %v\n", err.Error())
			}
		case msg, ok := <-consumer.Messages():
			if ok {
				// err := logic.Write(string(msg.Key), msg.Value)
				logmsg := "Write message to buffer, "
                logmsg = logmsg + "Topic: %s, Partition: %d, Offset: %d, Key: %v, Value: %s\n"
                logger.Printf(logmsg, msg.Topic, msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
				err := logic.buffWrite(string(msg.Key), msg.Value)
				if err != nil {
					logmsg := "Fail to write message to buffer: %v. "
					logmsg = logmsg + "Topic: %s, Partition: %d, Offset: %d, Key: %v, Value: %s\n"
					logger.Errorf(logmsg, err, msg.Topic, msg.Partition, msg.Offset, msg.Key, msg.Value)
				} else {
					consumer.MarkOffset(msg, "") // mark message as processed
				}

			}
		case <-after_1_minute:
			return nil
		}
	}
	return nil
}

func CloseConsumer(c *cluster.Consumer) {
	err := c.Close()
	util.ErrorHandler("Close Kafka consumer", err, func() error {
		return c.Close()
	})
}
